package aos.dijkstra;

import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Queue;

import aos.entity.Event;
import aos.entity.EventTable;
import aos.entity.Message;
import aos.entity.MessageType;
import aos.entity.RouteTable;
import aos.exception.AosException;
import aos.log.LogUtils;
import aos.socket.SocketHandler;
import aos.socket.SocketListener;

public class ProcessorNode implements Runnable {
	private int nodeId;
	private int logicTime;
	private int fatherNodeId;
	private boolean isActive;
	private boolean isRoot;
	private HashMap<Integer, Integer> childNodeId2ReqNum;
	private RouteTable routeTable;
	private EventTable eventTable;
	private SocketListener tcpServer;
	private HashMap<Integer, SocketHandler> id2Socket;
	private Queue<Message> messageQueue;

	public ProcessorNode(int port, String propertiesFile, String eventsFile)
			throws AosException {
		logicTime = 0;
		fatherNodeId = -1;
		isActive = false;
		isRoot = false;

		childNodeId2ReqNum = new HashMap<Integer, Integer>();
		routeTable = new RouteTable(propertiesFile);
		eventTable = new EventTable(eventsFile);
		id2Socket = new HashMap<Integer, SocketHandler>();
		messageQueue = new LinkedList<Message>();
		nodeId = routeTable.getLocalNodeId();

		tcpServer = new SocketListener(port, id2Socket, routeTable,
				messageQueue);
		new Thread(tcpServer).start();
	}

	@Override
	public void run() {
		try {
			runEvent();
		} catch (AosException e) {
			LogUtils.errorLog(e.getMessage());
			return;
		}
		while (true) {
			synchronized (messageQueue) {
				try {
					messageQueue.wait();
				} catch (InterruptedException e) {
					LogUtils.errorLog(e.getMessage());
				}
				while (!messageQueue.isEmpty()) {
					Message msg = messageQueue.poll();
					try {
						handleMessageReceive(msg);
					} catch (AosException e) {
						LogUtils.errorLog("handleMessageReceive failed: "
								+ msg.toString());
					}
				}
			}
		}
	}

	private void runEvent() throws AosException {
		Event event = null;
		while ((event = eventTable.getEvent(nodeId, logicTime)) != null) {
			handleEventHappen(event);
		}
		LogUtils.debugLog("no more event to run at logic time " + logicTime);
	}

	private void handleEventHappen(Event event) throws AosException {
		LogUtils.runLog("Handling event...  " + event.toString());
		LogUtils.infoLog("Event happens at logic time " + logicTime
				+ ". Event Info: " + event.toString());
		if (event.getEventType().equalsIgnoreCase("INIT")) {
			isActive = true;
			LogUtils.infoLog("Node " + nodeId
					+ " join the tree as root at logic time " + logicTime);
			LogUtils.keyLog("NodeId: " + event.getNodeId() + " LogicTime: "
					+ event.getLogicTime() + " EventType: "
					+ event.getEventType());
			isRoot = true;
			++logicTime;
		} else if (event.getEventType().equalsIgnoreCase("SEND")) {
			isActive = true;
			Message msg = new Message(nodeId, event.getOption(),
					event.getLogicTime(), MessageType.REQ);
			sendMessage(msg);
			if (childNodeId2ReqNum.containsKey(msg.getToId())) {
				childNodeId2ReqNum.put(msg.getToId(),
						childNodeId2ReqNum.get(msg.getToId()) + 1);
			} else {
				childNodeId2ReqNum.put(msg.getToId(), 1);
			}
			++logicTime;
		} else if (event.getEventType().equalsIgnoreCase("TICK")) {
			isActive = true;
			try {
				Thread.sleep(event.getOption());
			} catch (InterruptedException e) {
				LogUtils.errorLog(e.getMessage());
			}
			LogUtils.keyLog("NodeId: " + event.getNodeId() + " LogicTime: "
					+ event.getLogicTime() + " EventType: "
					+ event.getEventType() + " MilliSecond: "
					+ event.getOption());
			++logicTime;
		} else if (event.getEventType().equalsIgnoreCase("IDLE")) {
			isActive = false;
			LogUtils.keyLog("NodeId: " + event.getNodeId() + " LogicTime: "
					+ event.getLogicTime() + " EventType: "
					+ event.getEventType());
			++logicTime;
			tryToLeaveTree();
		}
	}

	private void handleMessageReceive(Message msg) throws AosException {
		LogUtils.runLog("Handling message...  " + msg.toString());
		logicTime = Math.max(logicTime, msg.getLogicTime() + 1);
		LogUtils.infoLog("Message receives at logic time " + logicTime
				+ ". Message Info: " + msg.toString());
		LogUtils.keyLog("NodeId: " + nodeId + " LogicTime: " + logicTime
				+ " EventType: RECV" + " FromNodeId: " + msg.getFromId()
				+ " MessageType: " + msg.getType());
		if (msg.getType() == MessageType.REQ) {
			isActive = true;
			if (fatherNodeId == -1 && !isRoot) {
				fatherNodeId = msg.getFromId();
				LogUtils.infoLog("nodeId " + nodeId
						+ " joins the tree with parentId " + fatherNodeId
						+ " at logic time " + logicTime);
			} else {
				LogUtils.infoLog("nodeId " + nodeId
						+ " is already in the tree with parentId "
						+ fatherNodeId + " at logic time " + logicTime);
				++logicTime;
				Message ackMsg = new Message(nodeId, msg.getFromId(),
						logicTime, MessageType.ACK);
				sendMessage(ackMsg);
				LogUtils.infoLog("Message sends at logic time " + logicTime
						+ ". Message Info: " + ackMsg.toString());
			}
			++logicTime;
		} else if (msg.getType() == MessageType.ACK) {
			childNodeId2ReqNum.put(msg.getFromId(),
					childNodeId2ReqNum.get(msg.getFromId()) - 1);
			++logicTime;
			tryToLeaveTree();
		}
		runEvent();
	}

	private void tryToLeaveTree() throws AosException {
		if (!isReallyIdle()) {
			LogUtils.debugLog("node is not really idle at logic time "
					+ logicTime);
			return;
		}
		if (fatherNodeId != -1) {
			Message ackMsg = new Message(nodeId, fatherNodeId, logicTime,
					MessageType.ACK);
			sendMessage(ackMsg);
			LogUtils.infoLog("Message sends at logic time " + logicTime
					+ ". Message Info: " + ackMsg.toString());
			LogUtils.infoLog("nodeId " + nodeId
					+ " leaves the tree with parentId " + fatherNodeId
					+ " at logic time " + logicTime);
			fatherNodeId = -1;
			++logicTime;
		} else {
			LogUtils.infoLog("nodeId " + nodeId
					+ " leaves the tree as root at logic time " + logicTime
					+ ", Dijkstra-Scholten termination detection finished");
		}
	}

	private boolean isReallyIdle() {
		if (isActive) {
			return false;
		}
		for (Integer value : childNodeId2ReqNum.values()) {
			if (value != 0) {
				return false;
			}
		}
		return true;
	}

	public void sendMessage(Message msg) throws AosException {
		synchronized (id2Socket) {
			SocketHandler socketHandler = null;
			socketHandler = id2Socket.get(msg.getToId());
			if (socketHandler == null) {
				Socket socket = null;
				try {
					socket = new Socket(routeTable.getIpAddress(msg.getToId()),
							GlobalVariables.port);
				} catch (IOException e) {
					e.printStackTrace();
					return;
				}
				socketHandler = new SocketHandler(socket, messageQueue);
				id2Socket.put(msg.getToId(), socketHandler);
				new Thread(socketHandler).start();
			}
			socketHandler.sendMessage(msg);
		}
		LogUtils.keyLog("NodeId: " + msg.getFromId() + " LogicTime: "
				+ msg.getLogicTime() + " EventType: SEND" + " ToNodeId: "
				+ msg.getToId() + " MessageType: " + msg.getType());
	}

	public static void main(String args[]) throws Exception {
		try {
			GlobalVariables.initialize("global.txt");
		} catch (Exception e) {

		}
		LogUtils.initialize(GlobalVariables.loglevel, InetAddress
				.getLocalHost().getHostName() + "_" + GlobalVariables.logfile);

		ProcessorNode node = new ProcessorNode(GlobalVariables.port,
				GlobalVariables.iptable, GlobalVariables.events);
		new Thread(node).start();
	}
}
